/// libpcap-dev libhiredis-dev libpthread 
#include "packet_redis.h"
#include <linux/if_ether.h>
#include <string.h>
#include "zlog.h"

zlog_category_t *zc;

ThreadSafeQueue queue;
void queue_init(ThreadSafeQueue *q, size_t size) {
    q->buffer = malloc(size);
    q->head = q->tail = 0;
    q->capacity = size;
    pthread_mutex_init(&q->lock, NULL);
    pthread_cond_init(&q->cond, NULL);
}

int enqueue(ThreadSafeQueue *q, const uint8_t *data, size_t len) {
    pthread_mutex_lock(&q->lock);
    
    // 空间检查（网页7的流量控制思想）
    while ((q->tail + len) % q->capacity == q->head) {
        pthread_cond_wait(&q->cond, &q->lock);
    }
    
    memcpy(q->buffer + q->tail, data, len);
    q->tail = (q->tail + len) % q->capacity;
    pthread_cond_signal(&q->cond);
    pthread_mutex_unlock(&q->lock);
    return len;
}

int dequeue(ThreadSafeQueue *q, uint8_t *buf, size_t max_len) {
    pthread_mutex_lock(&q->lock);
    while (q->head == q->tail) {
        pthread_cond_wait(&q->cond, &q->lock);
   }
    
    size_t avail = (q->tail > q->head) ? (q->tail - q->head) : (q->capacity - q->head);
    size_t copy_len = avail > max_len ? max_len : avail;
    
    memcpy(buf, q->buffer + q->head, copy_len);
    q->head = (q->head + copy_len) % q->capacity;
    pthread_cond_signal(&q->cond);
    pthread_mutex_unlock(&q->lock);
    return copy_len;
}


void packet_handler(u_char *user, const struct pcap_pkthdr *h, const u_char *bytes) {
    ThreadSafeQueue *q = (ThreadSafeQueue *)user;
    zlog_info(zc, "enqueue %d bytes", h->caplen);
    enqueue(q, bytes, h->caplen); // 原始数据入队
}

void *capture_thread(void *arg) {
    ThreadSafeQueue *q = (ThreadSafeQueue *)arg;
    char errbuf[PCAP_ERRBUF_SIZE];
    
    // 初始化抓包（网页8的设备配置）
    pcap_t *handle = pcap_open_live("eth0", BUFSIZ, 1, 1000, errbuf);

    ///  下面三行添加过滤器
    struct bpf_program fp;
    pcap_compile(handle, &fp, "tcp", 0, PCAP_NETMASK_UNKNOWN); // 过滤HTTP流量
    pcap_setfilter(handle, &fp);
    
    pcap_loop(handle, -1, packet_handler, (u_char *)q);
    pcap_close(handle);
    return NULL;
}

void *redis_thread(void *arg) {
    RedisConfig *cfg = (RedisConfig *)arg;
    
    // 连接Redis（网页2的认证处理）
    redisContext *c = redisConnect(cfg->host, cfg->port);
    if (c->err) {
        printf("Redis连接错误: %s\n", c->errstr);
        exit(1);
    }
    if (cfg->password) {
        redisCommand(c, "AUTH %s", cfg->password);
    }

    uint8_t packet_buf[ETH_FRAME_LEN];
    while(1) {
        int len = dequeue(&queue, packet_buf, sizeof(packet_buf));
        
        // 使用二进制协议发送（网页1的RESP协议）
        redisReply *reply = redisCommand(c, "PUBLISH %b %b", 
            REDIS_KEY, strlen(REDIS_KEY),
            packet_buf, len);
            
        if (reply->type == REDIS_REPLY_ERROR) {
            zlog_error(zc, "Redis错误: %s\n", reply->str);
        }
        freeReplyObject(reply);
	zlog_info(zc, "publish message");

    }
    redisFree(c);
    return NULL;
}

zlog_category_t *zlog_handler_init(const char *file_path)
{
    int rc = 0;

    rc = zlog_init(file_path);
    if(rc)
    {
	printf("zlog init failed!!!\n");
	return NULL;
    }

    zc = zlog_get_category("capture");
    if(!zc)
    {
	printf("zlog get category failed\n");
	return NULL;
    }
}

int main() {
    queue_init(&queue, QUEUE_SIZE);
    zc = zlog_handler_init("./log.conf");
    
    RedisConfig redis_cfg = {
        .host = "127.0.0.1",
        .port = 6379,
        .password = NULL
    };
    zlog_info(zc, "connect redis : [%s :%d]", redis_cfg.host, redis_cfg.port);

    pthread_t cap_tid, redis_tid;
    pthread_create(&cap_tid, NULL, capture_thread, &queue);
    pthread_create(&redis_tid, NULL, redis_thread, &redis_cfg);

    pthread_join(cap_tid, NULL);
    pthread_join(redis_tid, NULL);
    free(queue.buffer);
    zlog_fini();
    return 0;
}

